package com.burges.net.dataSet.annotation

import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.api.java.functions.FunctionAnnotation.{ForwardedFields, NonForwardedFields, ReadFields}
import org.apache.flink.api.scala.{ExecutionEnvironment, _}
import org.apache.flink.util.Collector

case class Person(name: String, age: Int)

object FieldsAnnotation {

  def main(args: Array[String]): Unit = {

    val env = ExecutionEnvironment.getExecutionEnvironment

    val dataSet1 = env.fromElements(("foo", 12), ("fun", 22))
    val dataSet2 = env.fromElements((12, 124), (23, 223))
    val dataSet3 = env.fromElements(Person("foo", 12), Person("fun", 22))

    /**
      * 转发注解的算子参数方式
      */
    // 指定Join函数
    val result = dataSet3.join(dataSet2).where("age").equalTo(1){
      (left, right, collector: Collector[(String, Double, Int)]) =>
            collector.collect(left.name, right._1+1, right._2)
            collector.collect("prefix" + left.name, right._1+2, right._2)
    }.withForwardedFieldsSecond("_2->_3")

    /**
      * Non-ForwardedFields注解使用实例代码
      */
    // 不转发第二个，其余字段转发到输出对象相同位置上
    @NonForwardedFields("_2")
    class MyMapper1 extends MapFunction[(String, Long, Int), (String, Long, Int)] {
      def map(input: (String, Long, Int)): (String, Long, Int) = {
        // 第一个和第三个字段不参与函数计算，第二个字段参与到函数计算过程中，并产生新的结果
        return (input._1, input._2 / 2, input._3 * 4)
      }
    }

    /**
      * 读取注解使用代码示例
      */
    // input的tuple的第一个字段和第三个字段参与函数的运算过程
    @ReadFields("_1; _2")
    class MyMapper extends MapFunction[(Int, Int, Double, Int), (Int, Long)]{
      def map(value: (Int, Int, Double, Int)): (Int, Double) = {
        if (value._1 == 42) {
          return (value._1, value._3)
        } else {
          return (value._2 + 10, value._3)
        }
      }
    }

    /**
      * 转发注解的函数注解实现方式
      */
    @ForwardedFields("_1->_2")
    class MyMapper2 extends MapFunction[(Int, Int, Double, Int), (Int, Long)]{
      def map(value: (Int, Int, Double, Int)): (Int, Double) = {
        if (value._1 == 42) {
          return (value._1, value._3)
        } else {
          return (value._2 + 10, value._3)
        }
      }
    }

  }

}
